package com.yzg.study.kafka.common.config;

import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.data.redis.cache.RedisCache;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.cache.RedisCacheWriter;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.util.ReflectionUtils;

import lombok.extern.slf4j.Slf4j;
import tk.mybatis.mapper.util.StringUtil;


@Slf4j
public class RedisTimeCacheManager extends RedisCacheManager implements ApplicationContextAware, InitializingBean {
	private ApplicationContext applicationContext;

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.applicationContext = applicationContext;
	}

	private Map<String, RedisCacheConfiguration> initialCacheConfiguration = new LinkedHashMap<>();

	public static final StringRedisSerializer STRING_SERIALIZER = new StringRedisSerializer();

	public static final GenericJackson2JsonRedisSerializer JACKSON_SERIALIZER = new GenericJackson2JsonRedisSerializer();

	public static final RedisSerializationContext.SerializationPair<String> STRING_PAIR = RedisSerializationContext.SerializationPair
			.fromSerializer(STRING_SERIALIZER);

	public static final RedisSerializationContext.SerializationPair<Object> FASTJSON_PAIR = RedisSerializationContext.SerializationPair
			.fromSerializer(JACKSON_SERIALIZER);

	public RedisTimeCacheManager(RedisCacheWriter cacheWriter, RedisCacheConfiguration defaultCacheConfiguration) {
		super(cacheWriter, defaultCacheConfiguration);
	}

	@Override
	protected Collection<RedisCache> loadCaches() {
		return initialCacheConfiguration.entrySet().stream()
				.map(entry -> super.createRedisCache(entry.getKey(), entry.getValue())).collect(Collectors.toList());
	}

	@Override
	public void afterPropertiesSet() {
		Stream.of(applicationContext.getBeanNamesForType(Object.class))
				.forEach(beanName -> add(applicationContext.getType(beanName)));
		super.afterPropertiesSet();
	}

	private void add(final Class<?> clazz) {
		ReflectionUtils.doWithMethods(clazz, method -> {
			ReflectionUtils.makeAccessible(method);
			method.getAnnotation(CacheExpire.class);
			CacheExpire cacheExpire = AnnotationUtils.findAnnotation(method, CacheExpire.class);
			if (cacheExpire == null) {
				return;
			}
			Cacheable cacheable = AnnotationUtils.findAnnotation(method, Cacheable.class);
			if (cacheable != null) {
				add(cacheable.cacheNames(), cacheExpire);
				return;
			}
		}, method -> null != AnnotationUtils.findAnnotation(method, CacheExpire.class));
	}

	private void add(String[] cacheNames, CacheExpire cacheExpire) {
		for (String cacheName : cacheNames) {
			if (StringUtil.isEmpty(cacheName)) {
				continue;
			}
			long expire = cacheExpire.expire();
			TimeType type = cacheExpire.type();
			log.info("缓存器名称: {}, 过期: {}", cacheName, expire);
			if (expire >= 0) {
				// 缓存配置
				RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
						.entryTtl(type.duration.apply(expire)).disableCachingNullValues()
						.serializeKeysWith(STRING_PAIR).serializeValuesWith(FASTJSON_PAIR);
				initialCacheConfiguration.put(cacheName, config);
			} else {
				log.warn("{} 使用默认的缓存过期时间.", cacheName);
			}
		}
	}
}
